package com.devshephard.framework;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.devshephard.business.service.UpdateUserInfo;
import com.devshephard.business.service.impl.UpdateUserInfoImpl;
import com.devshephard.entity.IUser;
import com.devshephard.mapper.IUserMapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.TimeUnit;

@Component
public class MqRunner implements ApplicationRunner, ApplicationContextAware {

   // private ApplicationContext applicationContext;

    @Autowired
    UpdateUserInfoImpl updateUserInfo;

    @Autowired
    IUserMapper iUserMapper;

    @Autowired
    ApplicationContext applicationContext;

    private static volatile Queue<MessageExt> waitQueue = new ArrayDeque<>();

    @Override
    public void run(ApplicationArguments args) throws Exception {
        //定义一个push消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");

        for (String beanDefinitionName : applicationContext.getBeanDefinitionNames()) {
            System.out.println(beanDefinitionName);
        }

        //指定nameserver
        consumer.setNamesrvAddr("192.168.81.130:9876");

        // 指定从第一条消息开始消费，指定消费offset
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 指定消费topic与tag *号表示全部tag
        consumer.subscribe("someTopic", "*");

        // 指定采用“广播模式”进行消费，默认为“集群模式”
        // consumer.setMessageModel(MessageModel.BROADCASTING);
        // 注册消息监听器 ,
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            // 一旦broker中有了其订阅的消息就会触发该方法的执行，
            // 其返回值为当前consumer消费的状态
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                try{
                    // 逐条消费消息
                    synchronized (waitQueue){
                        waitQueue.addAll(msgs);
                    }

                }catch (Exception e){
                    // 返回消费状态：消费成功
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                // 返回消费状态：消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            }
        });
        // 开启消费者消费
        consumer.start();

        new Thread(()->{
            while (true){
                try {
                    TimeUnit.SECONDS.sleep(300);
                    if(waitQueue.size() == 10){
                        synchronized (waitQueue){
                            System.out.println("UPDATE!!!!!");
                            for (int i = 0; i < 10; i++) {
                                waitQueue.poll();
                            }
                        }
                    }

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        System.out.println("Consumer Started");
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
